package com.ace.lab.starter.resolver;

import com.ace.lab.starter.utils.BeanFactoryHelper;
import com.google.common.base.Splitter;
import io.grpc.Attributes;
import io.grpc.EquivalentAddressGroup;
import io.grpc.NameResolver;
import io.grpc.internal.SharedResourceHolder;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.concurrent.GuardedBy;
import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class ZookeeperNameResolver extends NameResolver {

    private final Logger logger = LoggerFactory.getLogger(ZookeeperNameResolver.class);

    private final String path;
    private final Attributes params;
    private final SharedResourceHolder.Resource<ScheduledExecutorService> timerServiceResource;
    private final SharedResourceHolder.Resource<ExecutorService> sharedChannelExecutorResource;
    private Listener listener;
    private CuratorFramework curatorFramework = BeanFactoryHelper.getBeanFactory().getBean(CuratorFramework.class);
    private Set<String> zkNodes = new HashSet<>();
    private volatile boolean refreshing = false;
    private volatile boolean watching = false;

    public ZookeeperNameResolver(String path, Attributes params, SharedResourceHolder.Resource<ScheduledExecutorService> timerService,
                                 SharedResourceHolder.Resource<ExecutorService> sharedChannelExecutor) {
        this.path = path;
        this.params = params;
        this.timerServiceResource = timerService;
        this.sharedChannelExecutorResource = sharedChannelExecutor;
        checkProvider();
    }

    private void checkProvider() {
        ExecutorService executorService = SharedResourceHolder.get(sharedChannelExecutorResource);
        executorService.execute(() -> {
            List<String> childs = null;
            try {
                childs = curatorFramework.getChildren().forPath(path);
            } catch (Exception e) {
                e.printStackTrace();
            }
            if (childs == null) {
                logger.warn("there is no hosts found for service {}", path);
            } else {
                StringBuilder builder = new StringBuilder();
                builder.append(path);
                builder.append("\n");
                childs.forEach(child -> {
                    builder.append("├──── ");
                    builder.append(child);
                    builder.append("\n");
                });
                logger.info("found some host for service:\n{}", builder.toString());
            }
        });
    }

    @Override
    public String getServiceAuthority() {
        return "";
    }

    @Override
    public void start(Listener listener) {
        this.listener = listener;
        refresh();
    }

    @Override
    public void shutdown() {

    }

    @Override
    @GuardedBy("this")
    public void refresh() {
        if (refreshing) return;
        try {
            refreshing = true;
            List<String> childs = curatorFramework.getChildren().forPath(path);
            if (childs == null) {
                // Didn't find anything, retrying
                ScheduledExecutorService timerService = SharedResourceHolder.get(timerServiceResource);
                timerService.schedule(this::refresh, 30, TimeUnit.SECONDS);
                return;
            }
            zkNodes.addAll(childs);
            update();
            watch();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            refreshing = false;
        }
    }

    private void update() {
        List<EquivalentAddressGroup> servers = new ArrayList<>();
        zkNodes.forEach((String ip_port) -> {
            List<String> split = Splitter.on(":").splitToList(ip_port);
            InetSocketAddress socketAddress = new InetSocketAddress(split.get(0), Integer.parseInt(split.get(1)));
            EquivalentAddressGroup address = new EquivalentAddressGroup(socketAddress);
            servers.add(address);
        });
        listener.onAddresses(servers, params);
    }

    @GuardedBy("this")
    private void watch() throws Exception {
        if (watching) return;
        watching = true;
        PathChildrenCache childrenCache = new PathChildrenCache(curatorFramework, path, true);
        childrenCache.getListenable().addListener((client, event) -> {
            PathChildrenCacheEvent.Type type = event.getType();
            ChildData data;
            switch (type) {
                case INITIALIZED:
                    break;
                case CHILD_ADDED:
                    data = event.getData();
                    zkNodes.add(absolutePathToNode(data.getPath()));
                    update();
                    break;
                case CHILD_REMOVED:
                    data = event.getData();
                    zkNodes.remove(absolutePathToNode(data.getPath()));
                    update();
                    break;
                case CHILD_UPDATED:
                    break;
                case CONNECTION_SUSPENDED:
                    break;
                case CONNECTION_RECONNECTED:
                    break;
                case CONNECTION_LOST:
                    break;
                default:
                    break;
            }
        });
        childrenCache.start(PathChildrenCache.StartMode.BUILD_INITIAL_CACHE);
    }

    private String absolutePathToNode(String absolutePath) {
        return absolutePath.substring(path.length() + 1);
    }

}
